package com.dcloud.streamweb.taskmanage.controller;

import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.net.URLDecoder;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;

import com.dcloud.streamweb.constants.FlinkConstants;
import com.dcloud.streamweb.constants.SystemConstants;
import com.dcloud.streamweb.enums.DeployModeEnum;
import com.dcloud.streamweb.enums.JobTypeEnum;
import com.dcloud.streamweb.enums.SysErrorEnum;
import com.dcloud.streamweb.exception.BizException;
import com.dcloud.streamweb.taskmanage.entity.SavepointBackup;
import com.dcloud.streamweb.taskmanage.entity.dto.CheckPointDTO;
import com.dcloud.streamweb.taskmanage.entity.req.StartJobConfigReq;
import com.dcloud.streamweb.taskmanage.entity.req.StopJobConfigReq;
import com.dcloud.streamweb.taskmanage.entity.req.UpsertJobConfigReq;
import com.dcloud.streamweb.util.CliConfigUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.shiro.SecurityUtils;
import org.jeecg.common.api.vo.Result;
import org.jeecg.common.system.query.QueryGenerator;
import org.jeecg.common.system.vo.LoginUser;
import org.jeecg.common.util.oConvertUtils;
import com.dcloud.streamweb.taskmanage.entity.JobConfig;
import com.dcloud.streamweb.taskmanage.service.IJobConfigService;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import lombok.extern.slf4j.Slf4j;

import org.jeecgframework.poi.excel.ExcelImportUtil;
import org.jeecgframework.poi.excel.def.NormalExcelConstants;
import org.jeecgframework.poi.excel.entity.ExportParams;
import org.jeecgframework.poi.excel.entity.ImportParams;
import org.jeecgframework.poi.excel.view.JeecgEntityExcelView;
import org.jeecg.common.system.base.controller.JeecgController;
import org.quartz.CronExpression;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.multipart.MultipartFile;
import org.springframework.web.multipart.MultipartHttpServletRequest;
import org.springframework.web.servlet.ModelAndView;
import com.alibaba.fastjson.JSON;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import org.jeecg.common.aspect.annotation.AutoLog;
import org.apache.shiro.authz.annotation.RequiresPermissions;

/**
 * @Description: job_config
 * @Author: jeecg-boot
 * @Date: 2024-06-26
 * @Version: V1.0
 */
@Api(tags = "任务管理")
@RestController
@RequestMapping("/api/jobConfig/v1")
@Slf4j
public class JobConfigController extends JeecgController<JobConfig, IJobConfigService> {
    @Autowired
    private IJobConfigService jobConfigService;

    /**
     * 分页列表查询
     *
     * @param jobConfig
     * @param pageNo
     * @param pageSize
     * @param req
     * @return
     */
    //@AutoLog(value = "job_config-分页列表查询")
    @ApiOperation(value = "job_config-分页列表查询", notes = "job_config-分页列表查询")
    @GetMapping(value = "/list")
    public Result<IPage<JobConfig>> queryPageList(JobConfig jobConfig,
                                                  @RequestParam(name = "pageNo", defaultValue = "1") Integer pageNo,
                                                  @RequestParam(name = "pageSize", defaultValue = "10") Integer pageSize,
                                                  HttpServletRequest req) {
        QueryWrapper<JobConfig> queryWrapper = QueryGenerator.initQueryWrapper(jobConfig, req.getParameterMap());
        Page<JobConfig> page = new Page<JobConfig>(pageNo, pageSize);
        IPage<JobConfig> pageList = jobConfigService.page(page, queryWrapper);
        return Result.OK(pageList);
    }

    /**
     * 添加
     *
     * @param upsertJobConfigReq
     * @return
     */
    @AutoLog(value = "新增流任务")
    @ApiOperation(value = "新增流任务", notes = "新增流任务")
    @PostMapping(value = "/add")
    public Result<String> add(@RequestBody UpsertJobConfigReq upsertJobConfigReq) {

        Result result = this.checkUpsertJobConfigReq(upsertJobConfigReq);
        if (!result.isSuccess()) {
            return result;
        }

        jobConfigService.addJob(UpsertJobConfigReq.toDTO(upsertJobConfigReq));

        //jobConfigService.save(jobConfig);
        return Result.OK("添加成功！");
    }

    /**
     * @param upsertJobConfigReq
     * @Description: 参数检测
     * @return: void
     * @Author: shijf
     * @Date 2024/7/17 22:33
     */
    private Result checkUpsertJobConfigReq(UpsertJobConfigReq upsertJobConfigReq) {
        if (upsertJobConfigReq == null) {
            return Result.error("参数不能空");
        }
        if (StringUtils.isEmpty(upsertJobConfigReq.getJobName())) {
            return Result.error("任务名称不能空");
        }
        if (upsertJobConfigReq.getJobName().length() > 50) {
            return Result.error("任务名称不能超过50个字符");
        }
        if (!upsertJobConfigReq.getJobName().matches("[0-9A-Za-z_]*")) {
            return Result.error("任务名称仅能含数字,字母和下划线");
        }

        // jar需要校验参数
        if (JobTypeEnum.JAR.equals(upsertJobConfigReq.getJobType())) {

            if (StringUtils.isEmpty(upsertJobConfigReq.getCustomMainClass())) {
                return Result.error("主类不能为空");
            }

            if (StringUtils.isEmpty(upsertJobConfigReq.getCustomJarUrl())) {
                return Result.error("主类jar的不能为空");
            }
        }
        // sql配置需要校验的参数JobType=null是兼容之前配置
        if (JobTypeEnum.SQL_STREAMING.equals(upsertJobConfigReq.getJobType())
                || upsertJobConfigReq.getJobType() == null
                || JobTypeEnum.SQL_STREAMING.getCode() == upsertJobConfigReq.getJobType().intValue()) {
            if (StringUtils.isEmpty(upsertJobConfigReq.getFlinkSql())) {
                return Result.error("sql语句不能为空");
            }
            if (StringUtils.isNotEmpty(upsertJobConfigReq.getExtJarPath())) {
                String[] urls = upsertJobConfigReq.getExtJarPath().split(SystemConstants.LINE_FEED);
                for (String url : urls) {
                    if (StringUtils.isEmpty(url)) {
                        continue;
                    }
                }
            }
        }
        if (JobTypeEnum.SQL_BATCH.getCode() == upsertJobConfigReq.getJobType() && StringUtils
                .isNotEmpty(upsertJobConfigReq.getCron())) {
            if (!CronExpression.isValidExpression(upsertJobConfigReq.getCron())) {
                return Result.error("cron表达式不准确");
            }
        }

        //检测flink 检查点设置
        if (StringUtils.isNotEmpty(upsertJobConfigReq.getFlinkCheckpointConfig())) {
            CheckPointDTO checkPointParam = CliConfigUtil
                    .checkFlinkCheckPoint(upsertJobConfigReq.getFlinkCheckpointConfig());
            Result restResult = this.checkPointParam(checkPointParam);
            if (restResult != null && !restResult.isSuccess()) {
                return restResult;
            }
        }

        //如果指定的运行模式是yarn-per或者yarn-application，则需要校验flink运行配置
        if (DeployModeEnum.YARN_PER.name().equals(upsertJobConfigReq.getDeployMode()) ||
                DeployModeEnum.YARN_APPLICATION.name().equals(upsertJobConfigReq.getDeployMode())) {
            if (StringUtils.isEmpty(upsertJobConfigReq.getFlinkRunConfig())) {
                return Result.error("flink运行配置不能为空");
            }
            if (upsertJobConfigReq.getFlinkRunConfig().contains("-Dyarn.application.name=")) {
                return Result.error("请不要配置-Dyarn.application.name= 参数 ");
            }
        }

        return Result.ok();

    }


    private Result checkPointParam(CheckPointDTO checkPointParam) {
        if (checkPointParam == null) {
            return Result.ok();
        }
        if (StringUtils.isNotEmpty(checkPointParam.getCheckpointingMode())) {
            if (!(FlinkConstants.EXACTLY_ONCE.equalsIgnoreCase(checkPointParam.getCheckpointingMode())
                    || FlinkConstants.AT_LEAST_ONCE
                    .equalsIgnoreCase(checkPointParam.getCheckpointingMode()))) {
                return Result.error("checkpointingMode 参数必须是  AT_LEAST_ONCE 或者 EXACTLY_ONCE");
            }
        }
        if (StringUtils.isNotEmpty(checkPointParam.getExternalizedCheckpointCleanup())) {
            if (!(FlinkConstants.DELETE_ON_CANCELLATION
                    .equalsIgnoreCase(checkPointParam.getExternalizedCheckpointCleanup())
                    || FlinkConstants.RETAIN_ON_CANCELLATION
                    .equalsIgnoreCase(checkPointParam.getExternalizedCheckpointCleanup()))) {
                return Result.error(
                        "externalizedCheckpointCleanup 参数必须是DELETE_ON_CANCELLATION 或者 RETAIN_ON_CANCELLATION");
            }
        }
        return Result.ok();
    }


    /**
     * 添加
     *
     * @param startJobConfigReq
     * @return
     */
    @AutoLog(value = "任务启动")
    @ApiOperation(value = "任务启动", notes = "任务启动")
    @PostMapping(value = "/start")
    public Result<String> add(@RequestBody StartJobConfigReq startJobConfigReq) {
        try {
            LoginUser sysUser = (LoginUser) SecurityUtils.getSubject().getPrincipal();
            jobConfigService.startJob(startJobConfigReq, sysUser);
        } catch (Exception e) {
            e.printStackTrace();
            log.error("任务启动失败{}", e.getMessage());
            return Result.OK("任务启动失败！");
        }

        return Result.OK("任务启动成功！");
    }

    /**
     * 添加
     *
     * @param stopJobConfigReq
     * @return
     */
    @AutoLog(value = "任务停止")
    @ApiOperation(value = "任务停止", notes = "任务停止")
    @PostMapping(value = "/stop")
    public Result<String> stop(@RequestBody StopJobConfigReq stopJobConfigReq) {
        try {
            LoginUser sysUser = (LoginUser) SecurityUtils.getSubject().getPrincipal();
            jobConfigService.stop(stopJobConfigReq, sysUser);
        } catch (Exception e) {
            e.printStackTrace();
            log.error("任务停止失败{}", e.getMessage());
            return Result.OK("任务停止失败！");
        }

        return Result.OK("任务停止成功！");
    }

    @AutoLog(value = "触发保存点")
    @ApiOperation(value = "触发保存点", notes = "触发保存点")
    @GetMapping("/savepoint")
    public Result<String> savepoint(Long id) {
        try {
            jobConfigService.savepoint(id);

        } catch (BizException e) {
            log.warn("savepoint is error id={}", id, e);
            return Result.error(SysErrorEnum.SAVEPOINT_JOB_FAIL.getErrorMsg());
        } catch (Exception e) {
            log.error("savepoint is error id={}", id, e);
            return Result.error(SysErrorEnum.SAVEPOINT_JOB_FAIL.getErrorMsg());
        }
        return Result.ok("触发保存点成功");
    }


    @AutoLog(value = "保存点历史查询")
    @ApiOperation(value = "保存点历史查询", notes = "保存点历史查询")
    @GetMapping("/savepoint/history")
    public Result<?> savepointHitory(@RequestParam(name = "jobConfigId") Long jobConfigId) {
        List<SavepointBackup> list = jobConfigService.savepointHitory(jobConfigId);
        return Result.ok(list);
    }
}
